package cn.dmp.test

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object DmpMainServiceBak1 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getSimpleName)
      // 设置序列化方式， [rdd] [worker]
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      // 占用空间比较小
      .set("spark.rdd.compress", "true")
    val sc: SparkContext = new SparkContext(conf)

    val sqlContext: SQLContext = new SQLContext(sc)
    val parquet: DataFrame = sqlContext.read.parquet("logFile")
    val logRdd: RDD[Row] = parquet.rdd
    //logRdd.foreach(println)
    val splitRdd: RDD[Array[String]] = logRdd.map(row => {
      row.toString().split(",", -1)
    })
    //过滤
    val filterData: RDD[Array[String]] = splitRdd.filter(t => t.size == 85)
    filterData.cache()

    //第一问
    //FlowDistribution.accFlowDistribution(filterData, sqlContext)

    // 地域分布

    //          REQUESTMODE	PROCESSNODE	ISEFFECTIVE	ISBILLING	 ISBID	 ISWIN	ADORDERID  WinPrice    adpayment
    //24, 25     8,            35,           30,       31,       39,    42,     2          41           75
    //1   2      3             4              5         6         7      8      9          10           11
    val distributionInfo: RDD[(String, String, String, String, String, String, String, String, String, String, String)] = filterData.map(info => {
      (info(24), info(25), info(8), info(35), info(30), info(31), info(39), info(42), info(2), info(41), info(75))
    })
    distributionInfo.foreach(println)

    //按省分组
    val etlInfo: RDD[(String, List[Double])] = distributionInfo.map(info => {
      val requestNum: Int = if (info._3.equals("1") && info._4.toInt >= 1) 1 else 0
      val validNum: Int = if (info._3.equals("1") && info._4.toInt >= 2) 1 else 0
      val adRequestNum: Int = if (info._3.equals("1") && info._4.toInt == 3) 1 else 0
      val participationNum: Int = if (info._5.equals("1") && info._6.equals("1") && info._7.equals("1") && !info._8.equals("0")) 1 else 0
      val successfulsBidNum: Int = if (info._5.equals("1") && info._6.toInt >= 1 && info._8.equals("1")) 1 else 0
      val showNum: Int = if (info._3.equals("2") && info._5.toInt == 1) 1 else 0
      val clickNum: Int = if (info._3.equals("3") && info._5.toInt == 1) 1 else 0
      val dspConsume: Double = if (info._5.equals("1") && info._6.equals("1") && info._7.equals("1")) info._10.toDouble / 1000 else 0
      val dspCost: Double = if (info._5.equals("1") && info._6.equals("1") && info._7.equals("1")) info._11.toDouble / 1000 else 0
      (info._1, List[Double](requestNum, validNum, adRequestNum, participationNum, successfulsBidNum, showNum, clickNum, dspConsume, dspCost))
    })

    val resultPro: RDD[(String, List[Double])] = etlInfo.reduceByKey((list1, list2) => {
      list1.zip(list2).map(t => t._1 + t._2)
    })
    resultPro.foreach(println)

    //按省市分组
    val etlInfoCity: RDD[(String, List[Double])] = distributionInfo.map(info => {
      val requestNum: Int = if (info._3.equals("1") && info._4.toInt >= 1) 1 else 0
      val validNum: Int = if (info._3.equals("1") && info._4.toInt >= 2) 1 else 0
      val adRequestNum: Int = if (info._3.equals("1") && info._4.toInt == 3) 1 else 0
      val participationNum: Int = if (info._5.equals("1") && info._6.equals("1") && info._7.equals("1") && !info._8.equals("0")) 1 else 0
      val successfulsBidNum: Int = if (info._5.equals("1") && info._6.toInt >= 1 && info._8.equals("1")) 1 else 0
      val showNum: Int = if (info._3.equals("2") && info._5.toInt == 1) 1 else 0
      val clickNum: Int = if (info._3.equals("3") && info._5.toInt == 1) 1 else 0
      val dspConsume: Double = if (info._5.equals("1") && info._6.equals("1") && info._7.equals("1")) info._10.toDouble / 1000 else 0
      val dspCost: Double = if (info._5.equals("1") && info._6.equals("1") && info._7.equals("1")) info._11.toDouble / 1000 else 0
      (info._1, List[Double](requestNum, validNum, adRequestNum, participationNum, successfulsBidNum, showNum, clickNum, dspConsume, dspCost))
    })

    val resultProCity: RDD[(String, List[Double])] = etlInfoCity.reduceByKey((list1, list2) => {
      list1.zip(list2).map(t => t._1 + t._2)
    })
    resultProCity.foreach(println)

    sc.stop()
  }
}
